package com.cmsz.collection.service;

import com.cmsz.collection.util.DateUtil;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.time.Duration;
import java.time.Instant;
import java.util.*;

/**
 * @author liupeng
 * @date 2019/12/12
 * @description kafka监听实时推送服务
 */

@Component
@EnableScheduling
public class KafkaListenerPushInfoService {

    @Autowired
    InfoFilter infoFilter;

    @Qualifier("httpClientTemplate")
    @Autowired
    RestTemplate restTemplate;

    @Value("${infoPushUrl}")
    private String url;

    private static Logger logger = LoggerFactory.getLogger(KafkaListenerPushInfoService.class);

    private static Map<String,Object> scanMap = new HashMap<>();

    private static Map<String,List> pushMap = new HashMap<>();

    private static boolean isBatchPush = false;

    private static int count = 0;

//    @KafkaListener(topics = {"#{'${spring.kafka.listener.topics}'.split(',')}"},
//            containerFactory = "kafkaListenerContainerFactory1",
//            groupId = "${spring.kafka.listener.pushInfo.group-id}")
    public void listen(ConsumerRecord record, Acknowledgment ack){
        try {
            Map<String, String> infoMessage = new HashMap<>();
            JSONObject value = JSONObject.fromObject(record.value());
            String message = (String) value.get("message");    // 日志信息
            JSONObject fields = JSONObject.fromObject(value.get("fields")); // 获取fields字段信息，包括主机名和hostname
            String hostName = (String) fields.get("hostname");      // 采集主机ip
            String applicationName = (String) fields.get("applicationName");    // 采集应用名
            String time = (String) value.get("@timestamp");           // 采集时间
            String logType = record.topic();
            scanMap.put(hostName, time);
            // TODO 将json数据发送给接口层，目前没有接口,仅执行输出操作
            if (isPush(logType,applicationName)) {
                infoMessage.put(record.topic(), message);
                JSONObject json = JSONObject.fromObject(infoMessage);
                System.err.println("实时推送"+json);
                String returnResult = doService(json);
                infoMessage.clear();
                // 开发环境不提交偏移量，方便测试
//            ack.acknowledge(); //提交偏移量
            }
        }catch (Exception e){
            logger.error("日志推送服务出现异常，异常消息：",e.getMessage());
        }
    }

    /**
     * 批量推送，将topic的提交，交由注解自动提交
     * @param records  topic中监听到的数据
     */
        @KafkaListener(topics = {"#{'${spring.kafka.listener.topics}'.split(',')}"},
            containerFactory = "infoPushKafkaListenerContainerFactory",
            groupId = "${spring.kafka.listener.pushInfo.group-id}")
    public void listen(List<ConsumerRecord>records){
        try{
            // 测试通过之后删除
            Instant begin = Instant.now();
            List<Map> infoMessage = new ArrayList<>();
            records.forEach(record -> {

                Map<String,String> initRecord = initRecord(record);
                synchronized (this) {
                    scanMap.put(initRecord.get("hostName"), initRecord.get("time"));
                }
                if (isPush(initRecord.get("logType"),initRecord.get("appName"))){
                    Map map = new HashMap();
                    map.put(initRecord.get("logType"),initRecord.get("message"));
                    infoMessage.add(map);
                }
                // 当接口层发送一个新的请求时，立刻将list集合中的数据推送给接口层，然后开始保存新的需要推送数据
                if (isBatchPush && !infoMessage.isEmpty()){
                    JSONArray jsonArray = JSONArray.fromObject(infoMessage);
                    String service = doService(jsonArray);
                    isBatchPush = false;
                    infoMessage.clear();
                }
//                System.out.println(++count);
            });
            // 批量获取到的数据消费完后，推送一次日志数据给接口层
            if (!infoMessage.isEmpty()) {
                JSONArray jsonArray = JSONArray.fromObject(infoMessage);
                String service = doService(jsonArray);
                isBatchPush = false;
                infoMessage.clear();
            }
            // 测试通过之后删除
            Instant end = Instant.now();
            long millis = Duration.between(begin, end).toMillis();
            logger.info("批量获取到 " + records.size() + " 条数据，处理耗时 " + millis + "ms");

        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 单条数据发送给接口层
     * @param json 需要发送的日志消息，map格式
     * @return 返回的执行结果
     */
    private String doService(JSONObject json){
//        System.out.println(json);
        String returnResult = restTemplate.postForEntity(url, json, String.class).getBody();//将json数据发送到接口层
        return "SUCCESS";
    }

    /**
     * 批量数据发送到接口层
     * @param json 需要发送的批量日志信息，list格式
     * @return 返回的执行结果
     */
    private String doService(JSONArray json){
        // TODO 上线时删除掉输出语句
//        System.out.println(json);
        String returnResult = restTemplate.postForEntity(url, json, String.class).getBody();//将json数据发送到接口层
//        System.out.println(returnResult);
        return "SUCCESS";
    }


    /**
     * 定时任务监听filebeat是否宕机
     * 1. 获取kafka topic中的filebeat采集的主机 和采集时间，将主机ip作为key，采集时间作为value，暂存在map中
     * 2. 遍历map，获取每个ip的采集时间，与当前时间对比，如果时间超过10分钟，打印告警信息
     */
    @Scheduled(cron = "0 */10 * * * ?")
    public void fileBeatDetection(){
        logger.info("监控fileBeat运行状态");
        scanMap.forEach((k,v)->{
            String time = v.toString();
            Date timestamp = DateUtil.UTCTimeToLocalDate(time); // 采集日志时间
            Instant now = Instant.now();    // 当前时间
            if (timestamp!=null) {
                long timeLag = Duration.between(timestamp.toInstant(), now).toMillis();
                int minTimeLag = (int)timeLag / (60 * 1000);
                if (minTimeLag > 10){
                    logger.warn("主机ip："+k+" 已经超过10分钟没上传过日志，请检查是否宕机，最后上传日志时间为: " +DateUtil.UTCTimeToCurrentDateTime(time));
                }
            }else {
                logger.warn("当前主机:"+ k +"没采集过任何数据");
            }
        });
    }

    /**
     * 判断当前日志是否需要推送给前端页面
     * @param logType  日志类型
     * @param appName   当前日志所属的应用
     * @return  true/false
     */
    private synchronized boolean isPush(String logType,String appName){
        boolean isPush = false;

        List logTypeList = pushMap.get("logType") == null ? new ArrayList() : pushMap.get("logType");
        List appNameList = pushMap.get("appName") == null ? new ArrayList() : pushMap.get("appName");


        isPush = logTypeList.contains(logType) && appNameList.contains(appName);

        return isPush;

    }

    /**
     * 接收接口层发送过来的map数据
     * @param map   接口层发送过来的，包含需要推送的日志的信息
     */
    public synchronized static void setPushMap(Map<String,List> map){
        isBatchPush = true;
        pushMap.put("logType",map.get("loggerType")==null?new ArrayList<>():map.get("loggerType"));
        pushMap.put("appName",map.get("appName")==null?new ArrayList<>():map.get("appName"));
    }

    /**
     * 初始化record，将record中需要的消息，转换为map格式
     * @param record topic中一整条消息
     * @return 需要的数据map
     */
    private Map<String,String> initRecord(ConsumerRecord record){
        Map<String,String> initMap = new HashMap<>();
        JSONObject value = JSONObject.fromObject(record.value());
        String message = (String) value.get("message");    // 日志信息
        JSONObject fields = JSONObject.fromObject(value.get("fields")); // 获取fields字段信息，包括主机名和hostname
        String hostName = (String) fields.get("hostname");      // 采集主机ip
        String applicationName = (String) fields.get("applicationName");    // 采集应用名
        String logType = record.topic();
        String time = (String) value.get("@timestamp");           // 采集时间
        initMap.put("message",message);
        initMap.put("hostName",hostName);
        initMap.put("appName",applicationName);
        initMap.put("time",time);
        initMap.put("logType",logType);
        return initMap;
    }
}
